package com.kongxiang.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Time;
import com.kongxiang.storm.constant.Descid;

import java.sql.Timestamp;
import java.util.Date;
import java.util.Map;

/**
 * @version 1.0
 * @description:
 *   处理请求数据
 * @projectName: com.kongxiang.storm.bolt
 * @className: demo
 * @author:谭农春
 * @createTime:2018/9/29 19:24
 */
public class DealRequestBolt implements IRichBolt {

  OutputCollector collector;

  @Override
  public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
      this.collector = outputCollector;
  }

  @Override
  public void execute(Tuple tuple) {
    try {
      if (tuple.getSourceStreamId().equals(Descid.REQUEST_STREAM_ID.name()) ) {
        // EMIT 数据
        this.collector.emit(new Values(null,tuple.getValueByField("R_MSG"),System.currentTimeMillis()));
        collector.ack(tuple);
      }

    }catch (Exception e){
      collector.fail(tuple);
    }

  }

  @Override
  public void cleanup() {

  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declare(new Fields("id","message","createtime"));
  }

  @Override
  public Map<String, Object> getComponentConfiguration() {
    return null;
  }
}
